package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * @author Felix
 * @date 2023/9/24
 * 操作Hbase的工具类
 */
public class HbaseUtil {
    //获取hbase连接
    public static Connection getHbaseConnection() {
        try {
            Configuration conf = HBaseConfiguration.create();
            conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
            return ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            throw new RuntimeException();
        }
    }

    //获取支持异步操作Hbase的连接
    public static AsyncConnection getAsyncConnection() {
        try {
            Configuration conf = HBaseConfiguration.create();
            conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
            return ConnectionFactory.createAsyncConnection(conf).get();
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    //关闭hbase连接
    public static void closeHbaseConnection(Connection conn) {
        if (conn != null && !conn.isClosed()) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //关闭异步操作hbase的连接客户端
    public static void closeAsyncConnection(AsyncConnection asyncConnection){
        if(asyncConnection != null && !asyncConnection.isClosed()){
            try {
                asyncConnection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    /**
     * 从hbase中删除表
     *
     * @param conn      连接
     * @param nameSpace 表空间
     * @param tableName 表名
     */
    public static void dropHbaseTable(Connection conn, String nameSpace, String tableName) {
        try (Admin admin = conn.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            if (!admin.tableExists(tableNameObj)) {
                System.out.println("要删除的" + nameSpace + ":" + tableName + "不存在");
                return;
            }
            admin.disableTable(tableNameObj);
            admin.deleteTable(tableNameObj);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 在hbase中新建表
     *
     * @param conn      连接
     * @param nameSpace 表空间
     * @param tableName 表名
     * @param families  列族
     */
    public static void createHbaseTable(Connection conn, String nameSpace, String tableName, String... families) {
        try (Admin admin = conn.getAdmin()) {
            if (families.length < 1) {
                throw new IllegalArgumentException("在建表的时候至少需要传递一个列族");
            }
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            if (admin.tableExists(tableNameObj)) {
                System.out.println("要创建的" + nameSpace + ":" + tableName + "已存在");
                return;
            }
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            for (String family : families) {
                ColumnFamilyDescriptor columnFamilyDescriptor
                    = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }
            admin.createTable(tableDescriptorBuilder.build());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 从hbase表中删除一条数据
     *
     * @param conn      连接
     * @param nameSpace 表空间
     * @param tableName 表名
     * @param rowKey    rowkey
     */
    public static void delRow(Connection conn, String nameSpace, String tableName, String rowKey) throws IOException {
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        Table table = conn.getTable(tableNameObj);
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
        table.close();
    }

    /**
     * 向hbase中put一条数据
     *
     * @param conn         连接
     * @param nameSpace    表空间
     * @param tableName    表名
     * @param rowKey       rowkey
     * @param family       列族
     * @param columnNames  列
     * @param columnValues 列值
     */
    public static void putRow(Connection conn, String nameSpace, String tableName, String rowKey, String family, String[] columnNames, String[] columnValues) throws IOException {
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        Table table = conn.getTable(tableNameObj);
        Put put = new Put(Bytes.toBytes(rowKey));
        for (int i = 0; i < columnNames.length; i++) {
            String columnName = columnNames[i];
            String columnValue = columnValues[i];
            if (StringUtils.isNotEmpty(columnValue)) {
                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(columnName), Bytes.toBytes(columnValue));
            }
        }
        table.put(put);
        table.close();
    }

    //获取从hbase中读取字典数据创建动态表的建表语句
    public static String getBaseDicDDL() {
        return "CREATE TABLE base_dic (\n" +
            " dic_code string,\n" +
            " info ROW<dic_name string>,\n" +
            " PRIMARY KEY (dic_code) NOT ENFORCED\n" +
            ") " + getHbaseDDL("dim_base_dic");
    }

    //获取hbase连接器连接属性
    public static String getHbaseDDL(String tableName) {
        return "WITH (\n" +
            " 'connector' = 'hbase-2.2',\n" +
            " 'table-name' = '" + GmallConfig.HBASE_NAMESPACE + ":" + tableName + "',\n" +
            " 'zookeeper.quorum' = 'hadoop102,hadoop103,hadoop104:2181',\n" +
            " 'lookup.async' = 'true',\n" +
            " 'lookup.cache' = 'PARTIAL',\n" +
            " 'lookup.partial-cache.max-rows' = '500',\n" +
            " 'lookup.partial-cache.expire-after-write' = '1 hour',\n" +
            " 'lookup.partial-cache.expire-after-access' = '1 hour'\n" +
            ")";
    }

    /**
     * 根据rowkey从Hbase表中查询数据
     * @param conn          连接
     * @param nameSpace     表空间
     * @param tableName     表名
     * @param rowKey        主键
     * @return
     */
    public static JSONObject getDimInfoFromHbase(Connection conn,String nameSpace,String tableName,String rowKey){
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        try (Table table = conn.getTable(tableNameObj)){
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 以异步的方式  根据rowkey从hbase表中获取维度数据
     * @param asyncConnection      异步客户端连接
     * @param nameSpace            表空间
     * @param tableName            表名
     * @param rowKey               主键
     * @return
     */
    public static JSONObject getDimInfoFromHbaseByAsync(AsyncConnection asyncConnection,String nameSpace,String tableName,String rowKey){
        try {
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncConnection.getTable(tableNameObj);
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
    public static void main(String[] args) {
        System.out.println(getDimInfoFromHbase(getHbaseConnection(), GmallConfig.HBASE_NAMESPACE, "dim_base_trademark", "1"));
    }
}
